package com.why.websocketmqttstudy.why.mqtt.client;

import com.why.websocketmqttstudy.why.mqtt.callback.TopicMsgCallback;
import com.why.websocketmqttstudy.why.mqtt.config.MqttConfig;
import com.why.websocketmqttstudy.why.mqtt.handle.HandlerFactory;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;


import java.util.List;

/**
 * Mqtt客户端
 * @author why
 *
 */
public class MqttConsumerClient {

//    private static Logger logger = LoggerFactory.getLogger(MqttConsumerClient.class);

    public MqttConfig mqttCofig;
    
    private MqttClient client = null;

    private String[] msgTopic = null;
    
    private MqttConnectOptions options;
    
    private HandlerFactory handlerFactory;
    
    public MqttConsumerClient(HandlerFactory handlerFactory, MqttConfig mqttCofig)
    {
         this.mqttCofig = mqttCofig;
    	 List<String> topicList = mqttCofig.getTopicList();
         msgTopic = topicList.toArray(new String[topicList.size()]);
         
         options = new MqttConnectOptions();
         options.setUserName(mqttCofig.getUserName());
         options.setPassword(mqttCofig.getPassword().toCharArray());
         options.setCleanSession(false);
         // 设置超时时间(单位秒)
         options.setConnectionTimeout(mqttCofig.getTimeOut());
         // 设置会话心跳时间
         options.setKeepAliveInterval(mqttCofig.getKeepAlive());

         this.handlerFactory = handlerFactory;
    }
    
    /**
     * 用来连接服务器
     */
    public void connect() throws Exception 
    {
    	client = new MqttClient(mqttCofig.getHostUrl(), mqttCofig.getClientId(), new MemoryPersistence());
        client.setCallback(new TopicMsgCallback(this));
        client.connect(options);
        if (client.isConnected()) {
        	subscribe();
//        	logger.info("MQTT连接成功clientId:" + mqttCofig.getClientId() + ",client实例:" + client);
        }
    }
    
    /**
     * 重连
     * @throws MqttException 
     */
    public void reConnect()
    {
        System.out.println("重新尝试连接===============");
    	 while (true) 
    	 {
    		 if(null == client)
    			 return;
             if (!client.isConnected()) 
             {
                 try 
                 {
                	 client.connect(options);
                	 client.setCallback(new TopicMsgCallback(this));
                     if (client.isConnected()) 
                     {
                    	 subscribe();
                         break;
                     }
                 } 
                 catch (MqttException e) 
                 {
//                     logger.error("mqttClient 重连失败",e);
                 }
             } 
             else 
             {
//                 logger.info("mqttClient is null or connect");
             }
             
             try {
                 Thread.sleep(5000);
             } catch (InterruptedException e) {
                 e.printStackTrace();
             }
         }
    }
    
    /**
     * 订阅
     * @throws MqttException 
     */
    private void subscribe() throws MqttException 
    {
        //订阅消息
//        logger.info("============================================");
        if (null != msgTopic && msgTopic.length > 0) 
        {
            int[] qos = new int[msgTopic.length];
            for (int i = 0; i < msgTopic.length; i++) {
                qos[i] = 0;
            }
            System.out.println(msgTopic[0]);
            client.subscribe(msgTopic, qos);
        }
	}

	public HandlerFactory getHandlerFactory() {
		return handlerFactory;
	}
}
